package com.darcytech.debezium.core.rpc;

import java.io.IOException;
import java.net.ConnectException;
import java.util.concurrent.TimeUnit;

import com.darcytech.debezium.common.utils.NetUtils;
import com.darcytech.debezium.common.utils.ThreadUtils;
import com.darcytech.debezium.core.api.impl.ApiSender;
import com.darcytech.debezium.core.rpc.proto.Rpc;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;

import io.netty.handler.timeout.IdleStateHandler;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static com.darcytech.debezium.core.rpc.BootStrap.LOCAL_HOST;

public class RpcClient {

    private static final Logger LOGGER = LoggerFactory.getLogger(RpcClient.class);

    private final LeaderLatch leaderLatch;

    private final ApiSender apiSender;

    public RpcClient(LeaderLatch leaderLatch, ApiSender apiSender) {
        this.leaderLatch = leaderLatch;
        this.apiSender = apiSender;
    }

    public void start() {
        final LeaderLatch leaderLatch = this.leaderLatch;
        String lastRpcAddress = null;

        while (true) {
            try {
                String rpcAddress;
                try {
                    rpcAddress = leaderLatch.getLeader().getId();
                } catch (Exception e) {
                    LOGGER.error("obtain address error : {}", e.getMessage());
                    rpcAddress = lastRpcAddress;
                }
                if (rpcAddress == null || rpcAddress.equals("")) {
                    continue;
                }
                lastRpcAddress = rpcAddress;
                String[] addressAndPort = rpcAddress.split(":");

                String finalIp = addressAndPort[0];
                if (NetUtils.getIpAddress().equals(finalIp)) {
                    finalIp = LOCAL_HOST;
                }
                LOGGER.info("try connect to leader address is {}:{}", finalIp, addressAndPort[1]);
                connect(finalIp, Integer.parseInt(addressAndPort[1]));
                LOGGER.info("disconnect from  address is {}:{}", finalIp, addressAndPort[1]);
            } catch (Throwable t) {
                LOGGER.error("connect error : ", t);
                ThreadUtils.sleep(5000);
            }
        }
    }

    private void connect(String host, int port) throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new ProtobufVarint32FrameDecoder())
                                .addLast(new IdleStateHandler(0, 100, 0, TimeUnit.SECONDS))
                                .addLast(new ProtobufDecoder(Rpc.Resp.getDefaultInstance()))
                                .addLast(new ProtobufVarint32LengthFieldPrepender())
                                .addLast(new ProtobufEncoder())
                                .addLast(new RpcClientHandler(apiSender));
                    }
                });

        ChannelFuture f = b.connect(host, port).sync();
        LOGGER.info("connect to leader address  {}:{} successful", host, port);
        f.channel().closeFuture().sync();
        group.shutdownGracefully();
    }

}
